package com.topisv.tms.task.interfacetask;

import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.topisv.tms.entity.interfacetask.InterfaceTask;
import com.topisv.tms.entity.interfacetask.InterfaceTaskEnum;
import com.topisv.tms.service.interfacetask.InterfaceTaskProcessService;
import com.topisv.tms.service.interfacetask.InterfaceTaskService;
import com.topisv.tms.slf4j.MarkerConstant;

/**
 * 接口任务处理
 * 
 * @author <a href="mailto:yuanlianghe@nryuncang.com">yuanlianghe</a>
 */
public class InterfaceTaskProcessWorker extends Thread {
    private Logger logger = LoggerFactory.getLogger(getClass());
    
    private InterfaceTaskService taskService;
    private InterfaceTaskProcessService taskProcessService;
    
    private BlockingQueue<InterfaceTask> taskQueue = new ArrayBlockingQueue<InterfaceTask>(1000);
    private long SLEEP_MILLIS = 50l;
    
    
    public InterfaceTaskProcessWorker(String name, BlockingQueue<InterfaceTask> taskQueue,
            InterfaceTaskService taskService, InterfaceTaskProcessService taskProcessService) {
        setName(name);
        this.taskQueue = taskQueue;
        this.taskService = taskService;
        this.taskProcessService = taskProcessService;
    }
    

    @Override
    public void run() {
        while (true) {
            InterfaceTask task = null;
            
            try {
                task = taskQueue.take();
            } catch (InterruptedException e1) {
                logger.error("接口任务,从任务队列取数据异常。", e1);
            }
            
            if (task != null) {
                boolean isProcessing = false;
                try {
                    
                    isProcessing = InterfaceTaskProcessProvider.PROCESSING_MAP.containsKey(task.getId());
                    if (isProcessing) {
                        continue;
                    }
                    
                    InterfaceTaskProcessProvider.PROCESSING_MAP.putIfAbsent(task.getId(), 1);
                    
                    //检查是否已消费
                    boolean isProcessed = taskService.findIfProcessed(task.getId());
                    if (isProcessed) {
                        continue;
                    }
                    
                    long t0 = System.currentTimeMillis();
                    
                    //开始处理
                    taskProcessService.process(task);
                    
                    long t1 = System.currentTimeMillis();
                    
                    logger.info(MarkerConstant.MONITOR_MARKE, "接口任务处理: type:{},id:{}, 耗时:{}毫秒", task.getType(), task.getId(), t1 - t0);
                    
                    //处理成功， 更新任务信息
                    task.setUpdatedTime(new Date());
                    taskService.updateTask(task.getId(), InterfaceTaskEnum.STATUS_FINISH, InterfaceTaskEnum.WHO_NONE, "成功", new Date(), 1);
                } catch (Exception ex) {
                    String errorWho = null;
                    
                    if (ex instanceof SelfException) {
                        errorWho = InterfaceTaskEnum.WHO_SELF;
                    } else if (ex instanceof ThirdException) {
                        errorWho = InterfaceTaskEnum.WHO_THIRD;
                    } else {
                        errorWho = InterfaceTaskEnum.WHO_SELF;
                    }
                    
                    String errorMsg = StringUtils.substring(ex.getMessage(), 0 , 499);
                    //处理异常， 更新task信息
                    taskService.updateTask(task.getId(), InterfaceTaskEnum.STATUS_ERROR, errorWho, errorMsg, new Date(), 1);
                    
                } finally {
                    if (!isProcessing) {
                        if (InterfaceTaskProcessProvider.PROCESSING_MAP.containsKey(task.getId())) {
                            InterfaceTaskProcessProvider.PROCESSING_MAP.remove(task.getId());
                        }
                    }
                }
            }
            

            
            try {
                Thread.sleep(SLEEP_MILLIS);
            } catch (InterruptedException e) {
            }
        }
    }
}
